package com.orderpay_detect;

import com.orderpay_detect.bean.OrderEvent;
import com.orderpay_detect.bean.ReceiptEvent;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @Description: TODO QQ1667847363
 * @author: xiao kun tai
 * @date:2021/11/12 22:43
 *
 *  使用join的方式，只能匹配正确流
 * 两条流的订单交易匹配
 */
public class TxPayMatchByJoin {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //从文件中读取数据
        String orderPath = "OrderPayDetect/src/main/resources/OrderLog.csv";
        DataStream<String> orderStream = env.readTextFile(orderPath);

        String receiptPath = "OrderPayDetect/src/main/resources/ReceiptLog.csv";
        DataStream<String> receiptStream = env.readTextFile(receiptPath);

        //将读取的数据转换为POJO类型
        //读取订单支付事件数据
        DataStream<OrderEvent> orderEventStream = orderStream
                .map(line -> {
                    String[] fields = line.split(",");
                    return new OrderEvent(new Long(fields[0]), fields[1], fields[2], new Long(fields[3]));
                })
                .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<OrderEvent>() {
                    @Override
                    public long extractAscendingTimestamp(OrderEvent orderEvent) {
                        return orderEvent.getTimestamp() * 1000L;
                    }
                })
                .filter(data -> !"".equals(data.getTxId())); //交易id不为空，必须是pay事件


        //读取到账事件数据
        DataStream<ReceiptEvent> receiptEventStream = receiptStream
                .map(line -> {
                    String[] fields = line.split(",");
                    return new ReceiptEvent(fields[0], fields[1], new Long(fields[2]));
                })
                .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<ReceiptEvent>() {
                    @Override
                    public long extractAscendingTimestamp(ReceiptEvent receiptEvent) {
                        return receiptEvent.getTimestamp() * 1000L;
                    }
                });


        //区间连接两条流，得到匹配的数据
        SingleOutputStreamOperator<Tuple2<OrderEvent, ReceiptEvent>> resultStream = orderEventStream
                .keyBy(OrderEvent::getTxId)
                .intervalJoin(receiptEventStream.keyBy(ReceiptEvent::getTxId))
                .between(Time.seconds(-3), Time.seconds(5))   //-3,5区间范围
                .process(new TxPayMatchDetectByJoin());

        resultStream.print();

        env.execute("tx pay match by join job");
    }

    /**
     * 实现自定义ProcessJoinFunction函数
     */

    public  static class TxPayMatchDetectByJoin extends ProcessJoinFunction<OrderEvent,ReceiptEvent, Tuple2<OrderEvent,ReceiptEvent>>{
        @Override
        public void processElement(OrderEvent orderEvent, ReceiptEvent receiptEvent, Context context, Collector<Tuple2<OrderEvent, ReceiptEvent>> collector) throws Exception {
                collector.collect(new Tuple2<>(orderEvent,receiptEvent));
        }
    }
}
